package com.deepglint.wc;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import sun.tools.jar.Main;

/**
 * @author mj
 * @version 1.0
 * @date 2021-11-06 0:00
 *  并行度优先级：
 *  1、如果每个任务配置并行度，则以任务自己的并行度为准
 *  2、如果任务没有配置，则以env的并行度为准
 *  3、如果env没有配置，则以上传jar时配置的为准
 *  4、如果上传jar时没有配置，则以配置文件中的默认值为准
 */

// 流处理
public class StreamWorkCount {

    public static void main(String[] args) throws Exception {

        // 创建流处理执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 设置整个数据流的并行度，默认为5
//        env.setParallelism(1);
        // 全部禁用合并
//        env.disableOperatorChaining();

        // 从文件读取数据
//        String inputPath = "C:\\Users\\马军\\Desktop\\Idea-workspace\\flink\\src\\main\\resources\\hello.txt";
//
//        DataStream<String> inputDataStream = env.readTextFile(inputPath);

        // 使用parameter tool工具从启动参数中提取配置项
        ParameterTool tool = ParameterTool.fromArgs(args);
        String host = tool.get("host");
        int port = tool.getInt("port");

        // 从socket文本流读取数据
        // 如果没有设置共享组，则复用前面的共享组，如果前面没有slot共享组，则有一个"default"共享组
        DataStream<String> inputDataStream = env.socketTextStream(host, port);

        //基于数据流进行转换计算
        // 不同组的任务，一定会占用不同的slot
        // 所以可以为每一个任务设置一个共享组，将可以共享一个slot的任务分配到不同的slot上
        DataStream<Tuple2<String, Integer>> dateStream = inputDataStream.flatMap(new WorkCount.MyFlatMap())
                .slotSharingGroup("green")// 设置slot共享组
                .keyBy(0)
                .sum(1) //求和
                .setParallelism(2)// 设置并行度
                .slotSharingGroup("red") // 设置slot共享组
                .disableChaining() // 这一步操作，不参与前后的合并操作
                .startNewChain() // 不予前面的任务合并，可以于后面的任务合并，则将任务链分为两个部分
                .shuffle()// 重分区：完全随机
                ;

        dateStream.print().setParallelism(3);

        // 执行任务
        env.execute();
    }
}
